<?php

namespace app\controller;

use app\BaseController;

class Uplink extends BackendController{
    protected $noNeedLogin = '*';

    public function index()  {


        $conf = new \RdKafka\Conf();

        $conf->set('group.id', 'myConsumerGroup');

        $rk = new \RdKafka\Consumer($conf);
        $rk->addBrokers("60.255.187.176:19092");

        $topicConf = new \RdKafka\TopicConf();
        $topicConf->set('auto.commit.interval.ms', 100);
        $topicConf->set('offset.store.method', 'file');
        $topicConf->set('offset.store.path', sys_get_temp_dir());
        $topicConf->set('auto.offset.reset', 'smallest');

        $topic = $rk->newTopic("seemmoTopic", $topicConf);
        var_dump($topic);
// Start consuming partition 0
        $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);

        $message = $topic->consume(0, 1 * 10000);
        var_dump($message);var_dump(RD_KAFKA_RESP_ERR__TIMED_OUT);die();
        switch ($message->err) {
            case RD_KAFKA_RESP_ERR_NO_ERROR:
                //没有错误打印信息
                var_dump($message);
                break;
            case RD_KAFKA_RESP_ERR__PARTITION_EOF:
                echo "等待接收信息\n";
                break;
            case RD_KAFKA_RESP_ERR__TIMED_OUT:
                echo "超时\n";
                break;
            default:
                throw new \Exception($message->errstr(), $message->err);
                break;
        }
    }

}